package cn.doitedu.demo10;


import cn.doitedu.demo10.param_pojos.Model2Param;
import cn.doitedu.pojo.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/20
 * @Desc: 学大数据，上多易教育
 *    运算机模型类：
 *    支持的规则如下：
 *    // 年龄>25岁,且 性别 = male 的用户 ，规则生效后，做过 C行为3次以上，当他发生 K 行为，则命中规则
 *
 *    // 年龄<35岁,且 性别 = female ,且 活跃等级>2    的用户 ，规则生效后，做过 X行为5次以上，当他发生 P 行为，则命中规则
 *
 *    // 年龄<35岁,且 性别 = female ,且 活跃等级>2    的用户 ，规则生效后，做过 S行为2次以上，当他发生 P 行为，则命中规则
 *
 *
 *
 **/
public class RuleModel_2_Calculator implements RuleCalculator {

    Model2Param ruleParam;
    ValueState<Integer> cntState;
    Table htable;
    JSONObject message = new JSONObject();
    Roaring64Bitmap preSelectUsersBitmap;

    @Override
    public void init(RuntimeContext runtimeContext,
                     String ruleParamJson,
                     Roaring64Bitmap preSelectUsersBitmap,
                     Long webCalcEndTime) throws IOException {

        // 接收预圈选人群
        this.preSelectUsersBitmap = preSelectUsersBitmap;

        // 解析规则参数
        this.ruleParam = JSON.parseObject(ruleParamJson, Model2Param.class);

        // 申请flink状态
        this.cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleParam.getRule_id(), Integer.class));

        // 创建hbase客户端连接
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","doitedu:2181");
        Connection connection = ConnectionFactory.createConnection(conf);
        this.htable = connection.getTable(TableName.valueOf("user_profile"));

    }

    @Override
    public void calc(UserEvent eventBean, Collector<String> out) throws IOException {

        if(preSelectUsersBitmap.contains(eventBean.getUid())) {
            // 动态画像统计
            if (eventBean.getEvent_id().equals(ruleParam.getRealtime_profile().getEvent_id())) {
                cntState.update(cntState.value() == null ? 1 : cntState.value() + 1);
            }


            // 触发判断
            if (eventBean.getEvent_id().equals(ruleParam.getFire_condition().getEvent_id())) {

                // 1. 先判断动态画像条件是否已经满足
                if (cntState.value() != null && cntState.value() > ruleParam.getRealtime_profile().getMin_count()) {

                    message.put("rule_id", ruleParam.getRule_id());
                    message.put("user_id", eventBean.getUid());
                    message.put("hit_time", eventBean.getTimestamp());

                    out.collect(message.toJSONString());
                }
            }
        }
    }


}
